package org.argus.job.worker.core.actors;

import java.util.List;

import com.google.common.collect.Lists;

import lombok.extern.slf4j.Slf4j;
import org.argus.job.common.enums.ExecuteType;
import org.argus.job.common.enums.TimeExpressionType;
import org.argus.job.common.model.InstanceDetail;
import org.argus.job.common.request.ProcessorTrackerStatusReportReq;
import org.argus.job.common.request.ServerQueryInstanceStatusReq;
import org.argus.job.common.request.ServerScheduleJobReq;
import org.argus.job.common.request.ServerStopInstanceReq;
import org.argus.job.common.response.AskResponse;
import org.argus.job.worker.core.common.WorkerRuntime;
import org.argus.job.worker.core.tracker.manager.HeavyTaskTrackerManager;
import org.argus.job.worker.core.tracker.manager.LightTaskTrackerManager;
import org.argus.job.worker.core.tracker.task.TaskTracker;
import org.argus.job.worker.core.tracker.task.heavy.HeavyTaskTracker;
import org.argus.job.worker.core.tracker.task.light.LightTaskTracker;

/**
 * worker 的 master 节点，处理来自 server 的 jobInstance 请求和来自 worker 的task 请求
 *
 * @author tjq
 * @since 2020/3/17
 */
@Slf4j
public class TaskTrackerActor {

    private final WorkerRuntime workerRuntime;

    public TaskTrackerActor(WorkerRuntime workerRuntime) {
        this.workerRuntime = workerRuntime;
    }

    public void onReceiveServerScheduleJobReq(ServerScheduleJobReq req) {
        log.debug("[TaskTrackerActor] server schedule job by request: {}.", req);
        Long instanceId = req.getInstanceId();
        // 区分轻量级任务模型以及重量级任务模型
        // TODO 先全部处理成轻量级任务模型， 后续再确定是否支持MR等高阶应用
//        if (isLightweightTask(req)) {
            final LightTaskTracker taskTracker = LightTaskTrackerManager.getTaskTracker(instanceId);
            if (taskTracker != null) {
                log.warn("[TaskTrackerActor] LightTaskTracker({}) for instance(id={}) already exists.", taskTracker, instanceId);
                return;
            }
            // 判断是否已经 overload
            if (LightTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxLightweightTaskNum() * LightTaskTrackerManager.OVERLOAD_FACTOR) {
                // ignore this request
                log.warn("[TaskTrackerActor] this worker is overload,ignore this request(instanceId={}),current size = {}!",instanceId,LightTaskTrackerManager.currentTaskTrackerSize());
                return;
            }
            if (LightTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxLightweightTaskNum()) {
                log.warn("[TaskTrackerActor] this worker will be overload soon,current size = {}!",LightTaskTrackerManager.currentTaskTrackerSize());
            }
            // 创建轻量级任务
            LightTaskTrackerManager.atomicCreateTaskTracker(instanceId, ignore -> LightTaskTracker.create(req, workerRuntime));
//        } else {
//            HeavyTaskTracker taskTracker = HeavyTaskTrackerManager.getTaskTracker(instanceId);
//            if (taskTracker != null) {
//                log.warn("[TaskTrackerActor] HeavyTaskTracker({}) for instance(id={}) already exists.", taskTracker, instanceId);
//                return;
//            }
//            // 判断是否已经 overload
//            if (HeavyTaskTrackerManager.currentTaskTrackerSize() >= workerRuntime.getWorkerConfig().getMaxHeavyweightTaskNum()) {
//                // ignore this request
//                log.warn("[TaskTrackerActor] this worker is overload,ignore this request(instanceId={})! current size = {},", instanceId, HeavyTaskTrackerManager.currentTaskTrackerSize());
//                return;
//            }
//            // 原子创建，防止多实例的存在
//            HeavyTaskTrackerManager.atomicCreateTaskTracker(instanceId, ignore -> HeavyTaskTracker.create(req, workerRuntime));
//        }
    }

    /**
     * ProcessorTracker 心跳处理器
     */
    public void onReceiveProcessorTrackerStatusReportReq(ProcessorTrackerStatusReportReq req) {

        HeavyTaskTracker taskTracker = HeavyTaskTrackerManager.getTaskTracker(req.getInstanceId());
        if (taskTracker == null) {
            log.warn("[TaskTrackerActor] receive ProcessorTrackerStatusReportReq({}) but system can't find TaskTracker.", req);
            return;
        }
        taskTracker.receiveProcessorTrackerHeartbeat(req);
    }

    /**
     * 停止任务实例
     */
    public void onReceiveServerStopInstanceReq(ServerStopInstanceReq req) {

        log.info("[TaskTrackerActor] receive ServerStopInstanceReq({}).", req);
        HeavyTaskTracker heavyTaskTracker = HeavyTaskTrackerManager.getTaskTracker(req.getInstanceId());
        if (heavyTaskTracker != null) {
            heavyTaskTracker.stopTask();
            return;
        }
        LightTaskTracker lightTaskTracker = LightTaskTrackerManager.getTaskTracker(req.getInstanceId());
        if (lightTaskTracker != null) {
            lightTaskTracker.stopTask();
            return;
        }
        log.warn("[TaskTrackerActor] receive ServerStopInstanceReq({}) but system can't find TaskTracker.", req);
    }

    /**
     * 查询任务实例运行状态
     */
    public AskResponse onReceiveServerQueryInstanceStatusReq(ServerQueryInstanceStatusReq req) {
        AskResponse askResponse;
        TaskTracker taskTracker = HeavyTaskTrackerManager.getTaskTracker(req.getInstanceId());
        if (taskTracker == null && (taskTracker = LightTaskTrackerManager.getTaskTracker(req.getInstanceId())) == null) {
            log.warn("[TaskTrackerActor] receive ServerQueryInstanceStatusReq({}) but system can't find TaskTracker.", req);
            askResponse = AskResponse.failed("can't find TaskTracker");
        } else {
            InstanceDetail instanceDetail = taskTracker.fetchRunningStatus(req);
            askResponse = AskResponse.succeed(instanceDetail);
        }
        return askResponse;
    }


    private boolean isLightweightTask(ServerScheduleJobReq serverScheduleJobReq) {
        final ExecuteType executeType = serverScheduleJobReq.getExecuteType();
        // 非单机执行的一定不是
        if (executeType != ExecuteType.STANDALONE){
            return false;
        }
        TimeExpressionType timeExpressionType = serverScheduleJobReq.getTimeExpressionType();
        // 固定频率以及固定延迟的也一定不是
        return timeExpressionType != TimeExpressionType.FIXED_DELAY && timeExpressionType != TimeExpressionType.FIXED_RATE;
    }
}
